package com.rickie.pvuv.streaming;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import java.util.Properties;

public class StreamingKafkaSink {
    public static void main(String[] args) throws Exception {
        // 获取flink的运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // checkpoint
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        env.getCheckpointConfig().enableExternalizedCheckpoints(
                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
        );
        // 连接Socket获取输入的数据
        DataStreamSource<String> source = env.socketTextStream("localhost", 9900, "\n");
        // kafka config
        String brokerList = "centos-150:9092";
        String topic = "test_topic";
        Properties prop = new Properties();
        prop.setProperty("bootstrap.servers", brokerList);
        FlinkKafkaProducer<String> myProducer = new FlinkKafkaProducer<String>(
                brokerList, topic, new SimpleStringSchema());
        // 将数据写入kafka
        source.addSink(myProducer);
        source.print();
        // execute
        env.execute("StreamingKafkaSink");
    }
}
